package streaming.day02

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * 带状态更新
  * 需要设置检查点（用来存储数据，血统关系，配置信息等）
  */
object StreamingUpDateStateByKey {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
      .setAppName("StreamingUpdateStateByKey")
      .setMaster("local[*]")

    // 第二个参数为批次时间间隔，多长时间的数据集为一个批次，这个时间不能拍脑门设置，科学合理的设置,一般会比较短
    // 批次时间要参照spark ui -> total delay 时间来设置，批次时间最好要大于或者等于total delay
    val ssc = new StreamingContext(conf, Seconds(2)) // 2秒钟一个批次
    ssc.sparkContext.hadoopConfiguration

    // 设置checkpoint目录, 缓存历史批次数据
    /**
      * Metadata checkpointing - 将定义 streaming 计算的信息保存到容错存储（如 HDFS）中.这用于从运行 streaming 应用程序的 driver 的节点的故障中恢复（稍后详细讨论）. 元数据包括:
      *     Configuration - 用于创建流应用程序的配置.
      *     DStream operations - 定义 streaming 应用程序的 DStream 操作集.
      *     Incomplete batches - 批量的job 排队但尚未完成.
      *
      * Data checkpointing - 将生成的 RDD 保存到可靠的存储.[把当前批次计算出来的最终的单词次数结果存储到checkpoint目录中]
      *     这在一些将多个批次之间的数据进行组合的 状态 变换中是必需的.
      *     在这种转换中, 生成的 RDD 依赖于先前批次的 RDD, 这导致依赖链的长度随时间而增加.
      *     为了避免恢复时间的这种无限增加（与依赖关系链成比例）, 有状态转换的中间 RDD 会定期 checkpoint 到可靠的存储（例如 HDFS）以切断依赖关系链.
      */
    ssc.checkpoint("./ckpt")

    // 接收受数据
    val stream: ReceiverInputDStream[String] = ssc.socketTextStream("127.0.0.1", 44444)

    val wordcounts = stream.flatMap(_.split(" ")).map(word => (word, 1))
      .updateStateByKey(updateFunction _)

    wordcounts.print()


    ssc.start()
    ssc.awaitTermination()

  }

  /**
    * 带状态更新
    * @param newValues 当前批次某个单词出现的值序列集合（a => Seq[Int](1,1,1)）
    * @param runningCount 上一个批次跑出来的某个单词数量 a 之前的总的次数
    * @return
    */
  def updateFunction(newValues: Seq[Int], runningCount: Option[Int]): Option[Int] = {
    val newCount = newValues.sum + runningCount.getOrElse(0)
    Some(newCount)
  }

}
